package com.guchenbo.spark.stream

import java.util

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

import scala.util.Random

/**
 * @author guchenbo
 * @date 2022/4/15
 */
object KafkaProducer {
  def main(args: Array[String]) {

    val brokers = "localhost:9092"
    val topic = "spark"

    val words: List[String] = List("spark", "java", "scala", "hive", "flink")
    val len = words.length

    val props = new util.HashMap[String, Object]()
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")
    val producer = new KafkaProducer[String, String](props)
    // Send some messages
    while (true) {
      val size = Random.nextInt(10)
      for (_ <- 1 to size) {
        val str = words(Random.nextInt(len))
        println(str)
        val message = new ProducerRecord[String, String](topic, null, str)
        producer.send(message)
      }
      Thread.sleep(3000)
    }
  }
}
